package com.deep.example;


import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.atomic.AtomicInteger;

public class DataStreamJob {

    private static final Logger logger = LoggerFactory.getLogger(DataStreamJob.class);

    /* 下面示例对大于 500 和小于 500 的分别求和 */
    public static void main(String[] args) throws Exception {

        // 获取 flink 环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 添加 socket 文本流数据源
        //DataStreamSource<String> streamSource = env.fromElements("200", "100", "6000", "500", "2000", "300", "1500", "900");
        DataStreamSource<String> streamSource = env.socketTextStream("127.0.0.1", 7777);

        // 对大于 500 和小于 500 进行分组
        KeyedStream<String, String> stringKeyedStream = streamSource.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                int i = Integer.parseInt(s);
                return i > 500 ? "ge" : "lt";
            }
        });
        // 开 10 秒滚动窗口，每 10 秒为一批数据 【00:00:00 ~ 00:00:10）、【00:00:10 ~ 00:00:20）左闭右开区间
        WindowedStream<String, String, TimeWindow> windowedStream = stringKeyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        // 窗口处理函数，泛型 String, Integer, String, TimeWindow 依次对应 输入类型、输出类型、 KEY类型（即keyBy 返回的类型）, 窗口
        SingleOutputStreamOperator<Integer> outputStreamOperator = windowedStream.process(new ProcessWindowFunction<String, Integer, String, TimeWindow>() {
            /*
             * key: 分组的 key
             * context： 上下文信息
             * elements： 传过来的一批数据
             * out： 数据输出
             * */
            @Override
            public void process(String key, ProcessWindowFunction<String, Integer, String, TimeWindow>.Context context, Iterable<String> elements, Collector<Integer> out) throws Exception {
                System.out.println(key);
                AtomicInteger sum = new AtomicInteger();
                elements.forEach(item -> sum.addAndGet(Integer.parseInt(item)));
                out.collect(sum.get());
            }
        });
        // 输出
        outputStreamOperator.print();
        env.execute("分组求和");
    }
}

